package com.albert.WSMqtt;

import android.content.Context;
import android.os.Handler;
import android.os.HandlerThread;

import com.albert.tool.StringTool;

import org.eclipse.paho.android.service.MqttAndroidClient;
import org.eclipse.paho.client.mqttv3.DisconnectedBufferOptions;
import org.eclipse.paho.client.mqttv3.IMqttActionListener;
import org.eclipse.paho.client.mqttv3.IMqttDeliveryToken;
import org.eclipse.paho.client.mqttv3.IMqttToken;
import org.eclipse.paho.client.mqttv3.MqttCallbackExtended;
import org.eclipse.paho.client.mqttv3.MqttConnectOptions;
import org.eclipse.paho.client.mqttv3.MqttException;
import org.eclipse.paho.client.mqttv3.MqttMessage;

public class WSMqttService {
    private HandlerThread mqttThread;
    private Handler mqttHandler;
    private Context context;
    private WSMqttCallback callback;
    private String serverUri;
    private String clientId;
    private String subscripTopic;
    private String publishTopic;
    private String userName;
    private String password;
    private MqttAndroidClient mqttClient;

    public Boolean isConnected() {
        return mqttClient.isConnected();
    }

    /*
     * serverUri： mqtt服务器 如 tcp://192.168.43.190:1884
     * clientId： 连接端ID
     * subscripTopic: 注册topic
     * publishTopic:  发布topic
     * userName： 用户名 可选
     * password： 密码 可选
     * */
    public WSMqttService(Context context,
                         String serverUri,
                         String clientId,
                         String subscripTopic,
                         String publishTopic,
                         WSMqttCallback callback) {
        initParm(context, serverUri, clientId, subscripTopic, publishTopic, "", "", callback);
    }
    /*
     * serverUri： mqtt服务器 如 tcp://192.168.43.190:1884
     * clientId： 连接端ID
     * subscripTopic: 注册topic
     * publishTopic:  发布topic
     * userName： 用户名 可选
     * password： 密码 可选
     * */
    public WSMqttService(Context context,
                         String serverUri,
                         String clientId,
                         String subscripTopic,
                         String publishTopic,
                         String userName,
                         String password,
                         WSMqttCallback callback) {
        initParm(context, serverUri, clientId, subscripTopic, publishTopic, userName, password, callback);
    }

    public void initParm(Context context,
                         String serverUri,
                         String clientId,
                         String subscripTopic,
                         String publishTopic,
                         String userName,
                         String password,
                         WSMqttCallback callback) {
        this.context = context;
        this.serverUri = serverUri;
        this.clientId = clientId;
        this.subscripTopic = subscripTopic;
        this.publishTopic = publishTopic;
        this.userName = userName;
        this.password = password;
        this.callback = callback;
        if (StringTool.isBlank(serverUri) || StringTool.isBlank(clientId) ||
                StringTool.isBlank(subscripTopic) || StringTool.isBlank(publishTopic) ){
            return;
        }
        mqttThread = new HandlerThread("MqttThread");
        mqttThread.start();
        mqttHandler = new Handler(mqttThread.getLooper());
        mqttHandler.post(() -> { // 在这里执行耗时操作，避免阻塞主线程
            //初始化WebSocket
            initClient();
            connect();
        });
    }

    private void initClient() {
        clientId = clientId + System.currentTimeMillis();

        mqttClient = new MqttAndroidClient(context, serverUri, clientId);
        mqttClient.setCallback(new MqttCallbackExtended() {
            @Override
            public void connectComplete(boolean reconnect, String serverURI) {
                if (reconnect) {
                    if (callback != null) {
                        callback.connectComplete(reconnect, serverURI);
                    }
                    subscribeToTopic();
                } else {
                    callback.connectComplete(reconnect, serverURI);
                }
            }

            @Override
            public void connectionLost(Throwable cause) {

                if (callback != null) {
                    callback.connectionLost(cause);
                }
            }

            @Override
            public void messageArrived(String topic, MqttMessage message) throws Exception {

                if (callback != null) {
                    callback.messageArrived(topic, message);
                }
            }

            @Override
            public void deliveryComplete(IMqttDeliveryToken token) {
                if (callback != null) {
                    callback.deliveryComplete(token);
                }
            }
        });
    }

    private void connect() {
        MqttConnectOptions options = new MqttConnectOptions();
        options.setAutomaticReconnect(true);
        options.setCleanSession(true);
        options.setConnectionTimeout(30);
        options.setKeepAliveInterval(60);
        if (StringTool.isNotBlank(userName) && StringTool.isNotBlank(password)) {
            options.setUserName(userName);
            options.setPassword(password.toCharArray());
        }
//        options.setWill(topic, payload.getBytes(), qos, retained);

        try {
            mqttClient.connect(options, this, new IMqttActionListener() {
                @Override
                public void onSuccess(IMqttToken asyncActionToken) {
                    DisconnectedBufferOptions disconnectedBufferOptions = new DisconnectedBufferOptions();
                    disconnectedBufferOptions.setBufferEnabled(true);
                    disconnectedBufferOptions.setBufferSize(100);
                    disconnectedBufferOptions.setPersistBuffer(false);
                    disconnectedBufferOptions.setDeleteOldestMessages(false);
                    mqttClient.setBufferOpts(disconnectedBufferOptions);
                    subscribeToTopic();
                }

                @Override
                public void onFailure(IMqttToken asyncActionToken, Throwable exception) {
                    if (callback != null) {
                        callback.mqttConnectFailure(asyncActionToken, exception);
                    }
                }
            });
        } catch (MqttException e) {
            e.printStackTrace();
        }
    }

    public void subscribeToTopic() {
        try {
            int qos = 0;
            mqttClient.subscribe(subscripTopic, qos, this, new IMqttActionListener() {
                @Override
                public void onSuccess(IMqttToken asyncActionToken) {

                    if (callback != null) {
                        callback.mqttSubscribeSuccess(asyncActionToken);
                    }
                }
                @Override
                public void onFailure(IMqttToken asyncActionToken, Throwable exception) {
                    if (callback != null) {
                        callback.mqttSubscribeFailure(asyncActionToken, exception);
                    }
                }
            });
        } catch (MqttException e) {
            e.printStackTrace();
        }
    }
    private void unsubscribeTopic() {
        String[] topics = {subscripTopic};
        try {
            mqttClient.unsubscribe(topics, this, new IMqttActionListener() {
                @Override
                public void onSuccess(IMqttToken asyncActionToken) {
                    if (callback != null) {
                        callback.mqttUnsubscribeSuccess(asyncActionToken);
                    }
                }
                @Override
                public void onFailure(IMqttToken asyncActionToken, Throwable exception) {
                    if (callback != null) {
                        callback.mqttUnsubscribeFailure(asyncActionToken, exception);
                    }
                }
            });
        } catch (MqttException e) {
            e.printStackTrace();
        }
    }

    public void publishMessage(String jsonStr) {
        publishMessage(jsonStr, 0);
    }
    public void publishMessage(String jsonStr, int Qos) {
        if (mqttClient == null || !mqttClient.isConnected()) {
            return;
        }
//        mqttClient.getBufferedMessageCount()
        try {
            MqttMessage message = new MqttMessage();
            message.setQos(Qos);
            message.setPayload(jsonStr.getBytes());
            mqttClient.publish(publishTopic, message, this, new IMqttActionListener() {
                @Override
                public void onSuccess(IMqttToken asyncActionToken) {

                    if (callback != null) {
                        callback.mqttPublishSuccess(asyncActionToken);
                    }
                }

                @Override
                public void onFailure(IMqttToken asyncActionToken, Throwable exception) {
                    if (callback != null) {
                        callback.mqttPublishFailure(asyncActionToken, exception);
                    }
                }
            });
        } catch (MqttException e) {
            e.printStackTrace();
        }
    }

    public void disconnect() {
        mqttThread.quitSafely();
        if (mqttClient != null && mqttClient.isConnected()) {
            unsubscribeTopic();
            try {
                mqttClient.disconnect();
            } catch (MqttException e) {
                e.printStackTrace();
            }
        }
    }
}